package handler

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/go-redsync/redsync/v4"
	"github.com/go-redsync/redsync/v4/redis/goredis/v8"
	"github.com/streadway/amqp"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
	"gorm.io/gorm"
	"math/rand"
	"shop_srvs/inventory_srv/global"
	"shop_srvs/inventory_srv/model"
	"shop_srvs/inventory_srv/proto"
	"time"
)

type Inventory struct {
	proto.UnimplementedInventoryServer
}

func NewInventory() proto.InventoryServer {
	return &Inventory{}
}

func (i *Inventory) SetInv(ctx context.Context, req *proto.GoodsInvInfo) (*emptypb.Empty, error) {
	global.DB.Save(&model.Inventory{
		Goods:  req.GoodsId,
		Stocks: req.Num,
	})
	return &emptypb.Empty{}, nil
}
func (i *Inventory) InvDetail(ctx context.Context, req *proto.GoodsInvInfo) (*proto.GoodsInvInfo, error) {
	var inv model.Inventory
	if result := global.DB.Where(&model.Inventory{Goods: req.GoodsId}).First(&inv); result.RowsAffected == 0 {
		return nil, status.Errorf(codes.NotFound, "没有库存信息")
	}
	return &proto.GoodsInvInfo{
		GoodsId: inv.Goods,
		Num:     inv.Stocks,
	}, nil
}

func (i *Inventory) Sell(ctx context.Context, req *proto.SellInfo) (*proto.SellInfoResponse, error) {
	//扣减库存，本地事务
	//并发情况下，可能会出现超卖

	//方法3：redis锁
	pool := goredis.NewPool(global.RedisClient)
	rs := redsync.New(pool)
	//这个时候应该先查询表，然后确定这个订单是否已经扣减过库存，已经扣减过了就别扣减了
	var existingSellDetail model.StockSellDetail
	if result := global.DB.Where(&model.StockSellDetail{OrderSn: req.OrderSn}).First(&existingSellDetail); result.RowsAffected != 0 {
		return nil, status.Errorf(codes.AlreadyExists, "订单已处理")
	}

	tx := global.DB.Begin()
	var details []model.GoodsDetail
	for _, goodInfo := range req.GoodsInfo {
		var inv model.Inventory
		mutex := rs.NewMutex(fmt.Sprintf("goods_%d", goodInfo.GoodsId))
		if err := mutex.Lock(); err != nil {
			tx.Rollback()
			return nil, status.Errorf(codes.Internal, "获取redis分布式锁失败")
		}

		//有锁，判断库存是否充足
		tx.Where(&model.Inventory{Goods: goodInfo.GoodsId}).First(&inv)
		if inv.Stocks < goodInfo.Num {
			tx.Rollback()
			if err := i.unlock(mutex); err != nil {
				return nil, err
			}
			return &proto.SellInfoResponse{Msg: fmt.Sprintf("%d", goodInfo.GoodsId)},
				status.Errorf(codes.ResourceExhausted, "库存不足")
		}

		//扣减，会出现数据不一致问题		--锁，分布式锁
		inv.Stocks -= goodInfo.Num
		//方法2.2：乐观锁
		//这种写法有瑕疵，为什么？
		//零值 对于int类型来说 默认值是0 这种会被gorm给忽略掉(stock为1时)
		//if result := tx.Model(&model.Inventory{}).Select("Stocks", "Version").Where("goods=? and version=?",
		//	goodInfo.GoodsId, inv.Version).Updates(model.Inventory{Stocks: inv.Stocks,
		//	Version: inv.Version + 1}); result.RowsAffected == 0 {
		//	zap.S().Info("库存扣减失败")
		//} else { //不断重试
		//	zap.S().Info("库存扣减成功")
		//	break
		//}
		//if result := tx.Model(&model.Inventory{}).Select("Stocks", "Version").
		//	Where("goods=? and version=? and stocks>?", goodInfo.GoodsId, inv.Version, goodInfo.Num).
		//	Updates(map[string]any{
		//		"stocks":  gorm.Expr("`stocks - ?`", goodInfo.Num),
		//		"version": gorm.Expr("`version + ?`", 1),
		//	}); result.RowsAffected == 0 {
		//	zap.S().Info("库存扣减失败")
		//} else { //不断重试
		//	zap.S().Info("库存扣减成功")
		//}
		tx.Save(&inv)
		if err := i.unlock(mutex); err != nil {
			return nil, err
		}
		details = append(details, model.GoodsDetail{
			Goods: goodInfo.GoodsId,
			Num:   goodInfo.Num,
		})
	}

	sellDetail := model.StockSellDetail{
		OrderSn: req.OrderSn,
		Status:  1,
		Detail:  details,
	}
	//创建售卖详细表
	if res := tx.Create(&sellDetail); res.RowsAffected == 0 {
		tx.Rollback()
		return nil, status.Errorf(codes.Internal, "保存库存扣减历史失败")
	}

	tx.Commit() //手动提交
	return nil, nil
}
func (i *Inventory) Sell1(ctx context.Context, req *proto.SellInfo) (*proto.SellInfoResponse, error) {
	pool := goredis.NewPool(global.RedisClient)
	rs := redsync.New(pool)

	var existingSellDetail model.StockSellDetail
	if result := global.DB.Where(&model.StockSellDetail{OrderSn: req.OrderSn}).First(&existingSellDetail); result.RowsAffected != 0 {
		return nil, status.Errorf(codes.AlreadyExists, "订单已处理")
	}

	tx := global.DB.Begin()
	for {
		mutex := rs.NewMutex(fmt.Sprintf("goods_%s", req.OrderSn))
		if err := mutex.Lock(); err != nil {
			i := time.Duration(rand.Int()) * time.Millisecond
			time.Sleep(i)
			continue
		}
		var details []model.GoodsDetail
		for _, goodInfo := range req.GoodsInfo {
			var inv model.Inventory

			//有锁，判断库存是否充足
			tx.Where(&model.Inventory{Goods: goodInfo.GoodsId}).First(&inv)
			if inv.Stocks < goodInfo.Num {
				tx.Rollback()
				if err := i.unlock(mutex); err != nil {
					return nil, err
				}
				return &proto.SellInfoResponse{Msg: fmt.Sprintf("%d", goodInfo.GoodsId)},
					status.Errorf(codes.ResourceExhausted, "库存不足")
			}

			inv.Stocks -= goodInfo.Num
			tx.Save(&inv)
			if err := i.unlock(mutex); err != nil {
				return nil, err
			}
			details = append(details, model.GoodsDetail{
				Goods: goodInfo.GoodsId,
				Num:   goodInfo.Num,
			})
		}

		sellDetail := model.StockSellDetail{
			OrderSn: req.OrderSn,
			Status:  1,
			Detail:  details,
		}
		//创建售卖详细表
		if res := tx.Create(&sellDetail); res.RowsAffected == 0 {
			tx.Rollback()
			return nil, status.Errorf(codes.Internal, "保存库存扣减历史失败")
		}
		tx.Commit() //手动提交
		return nil, nil
	}

}
func (i *Inventory) unlock(mutex *redsync.Mutex) error {
	if ok, err := mutex.Unlock(); !ok || err != nil {
		return status.Errorf(codes.Internal, "释放redis分布式锁异常")
	}
	return nil
}
func (i *Inventory) Reback(ctx context.Context, req *proto.SellInfo) (*emptypb.Empty, error) {
	tx := global.DB.Begin()
	for _, goodInfo := range req.GoodsInfo {
		var inv model.Inventory
		if result := global.DB.Where(&model.Inventory{Goods: goodInfo.GoodsId}).First(&inv); result.RowsAffected == 0 {
			tx.Rollback()
			return nil, status.Errorf(codes.InvalidArgument, "没有库存信息")
		}
		//归还，会出现数据不一致问题		--锁，分布式锁
		inv.Stocks += goodInfo.Num
		tx.Save(&inv)
	}
	tx.Commit() //手动提交
	return &emptypb.Empty{}, nil
}

func AutoReback1(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
	type OrderInfo struct {
		OrderSn string
	}
	for i := range msgs {
		//既然是归还库存，那么应该具体的知道每件商品应该归还多少
		//接口应该确保幂等性，不能因为消息的重复发送导致一个订单的库存归还多次，没有扣减的库存你别归还
		//新建一张表，这张表记录了详细的订单扣减细节，以及归还细节
		var orderInfo OrderInfo
		_ = json.Unmarshal(msgs[i].Body, &orderInfo)

		//将inv的库存加回去，将selldetail的status设置为2，要在事务中进行
		tx := global.DB.Begin()
		var sellDetail model.StockSellDetail
		if res := tx.Model(&model.StockSellDetail{}).Where(&model.StockSellDetail{
			OrderSn: orderInfo.OrderSn,
			Status:  1,
		}).First(&sellDetail); res.RowsAffected == 0 { //没有要清除的selldetail
			return consumer.ConsumeSuccess, nil
		}

		//如果查询到那么逐个归还库存
		for _, orderGood := range sellDetail.Detail {
			if res := tx.Model(&model.Inventory{}).Where(&model.Inventory{Goods: orderGood.Goods}).
				Update("stocks", gorm.Expr("stocks+?", orderGood.Num)); res.RowsAffected == 0 {
				tx.Rollback()
				return consumer.ConsumeRetryLater, nil
			}
		}

		if result := tx.Model(&model.StockSellDetail{}).Where(&model.StockSellDetail{OrderSn: orderInfo.OrderSn}).
			Update("status", 2); result.RowsAffected == 0 {
			tx.Rollback()
			return consumer.ConsumeRetryLater, nil
		}
		tx.Commit()
		return consumer.ConsumeSuccess, nil
	}
	return consumer.ConsumeSuccess, nil
}
func AutoReback(msgs <-chan amqp.Delivery) {
	type OrderInfo struct {
		OrderSn string
	}

	//既然是归还库存，那么应该具体的知道每件商品应该归还多少
	//接口应该确保幂等性，不能因为消息的重复发送导致一个订单的库存归还多次，没有扣减的库存你别归还
	//新建一张表，这张表记录了详细的订单扣减细节，以及归还细节
	for msg := range msgs {
		var orderInfo OrderInfo
		_ = json.Unmarshal(msg.Body, &orderInfo)

		//将inv的库存加回去，将selldetail的status设置为2，要在事务中进行
		tx := global.DB.Begin()
		defer tx.Rollback()
		var sellDetail model.StockSellDetail
		if res := tx.Model(&model.StockSellDetail{}).Where(&model.StockSellDetail{
			OrderSn: orderInfo.OrderSn,
			Status:  1,
		}).First(&sellDetail); res.RowsAffected == 0 { //没有要清除的selldetail
			//如果为true表示确认所有未确认的消息
			//如果为false表示确认当前消息
			msg.Ack(false)
			continue
		}

		//如果查询到那么逐个归还库存
		for _, orderGood := range sellDetail.Detail {
			if res := tx.Model(&model.Inventory{}).Where(&model.Inventory{Goods: orderGood.Goods}).
				Update("stocks", gorm.Expr("stocks+?", orderGood.Num)); res.RowsAffected == 0 {
				tx.Rollback()
				msg.Reject(true)
				continue
			}
		}
		if result := tx.Model(&model.StockSellDetail{}).Where(&model.StockSellDetail{OrderSn: orderInfo.OrderSn}).
			Update("status", 2); result.RowsAffected == 0 {
			tx.Rollback()
			msg.Reject(true)
			continue
		}

		tx.Commit()
		//如果为true表示确认所有未确认的消息
		//如果为false表示确认当前消息
		msg.Ack(false)
	}
}
